package nettyRPC.proxy;


import nettyRPC.client.AsyncRPCCallback;
import nettyRPC.protocol.RpcRequest;
import nettyRPC.protocol.RpcResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @program: NettyRpc-master
 * @description
 * @author: tkk fendoukaoziji
 * @create: 2019-08-12 19:17
 **/
public class RpcFuture  implements Future<Object> {
    private static final Logger logger=LoggerFactory.getLogger(RpcFuture.class);

    private Sync sync;
    private RpcRequest request;
    private RpcResponse response;
    private long startTime;
    private long responseTimeThreshold=5000;

    private List<AsyncRPCCallback> pendingCallBack=new ArrayList<AsyncRPCCallback>();
    private ReentrantLock lock=new ReentrantLock();

    public RpcFuture(RpcRequest request) {
        this.sync=new Sync();
        this.request = request;
        this.startTime=System.currentTimeMillis();
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isDone() {
        return sync.isDone();
    }

    @Override
    public Object get() throws InterruptedException, ExecutionException {
        sync.acquire(-1);
        if(this.response!=null){
            return this.response.getResult();
        }else {
            return null;
        }
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        boolean success = sync.tryAcquireNanos(-1, unit.toNanos(timeout));
        if(success){
            if(this.response!=null){
                return this.response.getResult();
            }else {
                return null;
            }
        }else{
            throw new RuntimeException("TimeOut exception Request .id:>>"+this.request.getRequestId()
            +"/Request class name:>>"+this.request.getClassName()+ ". Request method: " + this.request.getMethodName());
        }

    }

    public void done(RpcResponse response){
        this.response=response;
        sync.release(1);
        invokeCallbacks();
        long responseTime=System.currentTimeMillis()-startTime;
        if(responseTime>this.responseTimeThreshold){
            logger.warn("Service response time is too slow. Request id = " + response.getRequestId() +
                    ". Response Time = " + responseTime + "ms");
        }
    }

    private void invokeCallbacks(){
        lock.lock();
        try {
            for (final AsyncRPCCallback callback    :pendingCallBack ) {
                runCallBack(callback);//这个方法会抛出异常 捕获yixia
            }
        } finally {
            lock.unlock();
        }

    }

    public RpcFuture addCallBack(AsyncRPCCallback callBack){
        lock.lock();
        try {
            if(isDone()){
                runCallBack(callBack);
            }else{
                this.pendingCallBack.add(callBack);
            }
        } finally {
            lock.unlock();
        }
        return this;
    }






    private void runCallBack(final AsyncRPCCallback callback){
        final RpcResponse response=this.response;
        MyRpcClient.submit(new Runnable() {
            @Override
            public void run() {
                if(!response.isError()){
                    callback.success(response.getResult());
                }else {
                    callback.fail(new RuntimeException("Response error", new Throwable(response.getError())));
                }
            }
        });
    }


    static class Sync extends AbstractQueuedSynchronizer{
        private static final long serialVerisonUID=1L;

        private final int done=1;
        private final  int pending=0;

        @Override
        protected boolean tryAcquire(int arg) {
            return getState() == done;
        }

        @Override
        protected boolean tryRelease(int arg){
            if(getState()==pending){
                if (compareAndSetState(pending, done)) {
                    return true;
                }else {
                    return false;
                }
            }else{
                return  true;
            }
        }

        public boolean isDone(){
            getState();
            return getState()==done;
        }

    }

}
























